package com.pengheng.transformation.base;

import com.pengheng.bean.WaterSensor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 映射方式实现基本转换算子
 */
public class TransMap {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<WaterSensor> stream = env.fromElements(
                new WaterSensor("sensor1", 1L, 1),
                new WaterSensor("sensor2", 2L, 2)
        );

//        stream.map(new MapFunction<WaterSensor, String>() {
//            @Override
//            public String map(WaterSensor waterSensor) throws Exception {
//                return waterSensor.id;
//            }
//        }).print();

        stream.map(new WaterSensorMap()).print();

        env.execute();

    }

    private static class WaterSensorMap implements MapFunction<WaterSensor, String> {
        @Override
        public String map(WaterSensor waterSensor) throws Exception {
            return waterSensor.id;
        }
    }

}
